#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <stdint.h>
#include <stdbool.h>
#include <errno.h>
#include <assert.h>

#include "event_msgqueue.h"
#include "../net/net.h"
#include <glog/logging.h>

//pthread_mutex_t request_count_mutex = PTHREAD_MUTEX_INITIALIZER;
long request_count = 0;

struct circqueue 
{
    volatile unsigned int head;
    volatile unsigned int tail;
    unsigned int array_elements;
    void **entries;
};

struct event_msgqueue 
{
    int queue_lwm;
    int push_fd;
    int pop_fd;
    int unlock_between_callbacks;
    bool enable_lock;

    struct event_base* evbase;
    struct event queue_ev;

    pthread_mutex_t lock;
    void (*callback)(struct event_base*, void *, void *);
    void *cbarg;
    struct circqueue *queue;
};

static struct circqueue *circqueue_new(unsigned int size) 
{
    struct circqueue *cq;

    cq = (struct circqueue*)calloc(1, sizeof(struct circqueue));
    if (cq == NULL){
        return NULL;
	}

	cq->array_elements = size;

    cq->entries = (void**)malloc(sizeof(void*) * cq->array_elements);
    if (cq->entries == NULL) {
        free(cq);
        return NULL;
    }

    return cq;
}

static void circqueue_destroy(struct circqueue *cq) 
{
    if (cq != NULL) {
        if (cq->entries != NULL) {
            free(cq->entries);
        }
        free(cq);
    }
}

static int circqueue_get_length(struct circqueue *cq) 
{
    return (cq->tail - cq->head);
}

static int circqueue_push_tail(struct circqueue *cq, void *elem) 
{
    int len = cq->array_elements - (cq->tail - cq->head);

    if (len == 0) {
		LOG(INFO)<<"queue full!!!";
        return -1;
    }

    int idx = cq->tail & (cq->array_elements - 1);
    cq->entries[idx] = elem;
    cq->tail += 1;
    
    __sync_fetch_and_add(&request_count, 1);

    return (cq->tail - cq->head);
}

static void *circqueue_pop_head(struct circqueue *cq) 
{
    void *data;

    if (cq->head == cq->tail){
        return NULL;
	}

    int idx = cq->head & (cq->array_elements - 1);
    data = cq->entries[idx];
    cq->head += 1;

    return data;
}

static void msgqueue_pop(int fd, short flags, void *arg) 
{
    struct event_msgqueue *msgq = (struct event_msgqueue*)arg;
    char buf[64];
    void *qdata = NULL;

    int ret = read(fd, buf, sizeof(buf));
    if (ret <= 0) {
        return;
    }

    if (msgq->enable_lock){
        pthread_mutex_lock(&msgq->lock);
	}

    while ((qdata = circqueue_pop_head(msgq->queue)) != NULL) {
        if (msgq->enable_lock && msgq->unlock_between_callbacks){
            pthread_mutex_unlock(&msgq->lock);
		}

        msgq->callback(msgq->evbase, qdata, msgq->cbarg);
        __sync_fetch_and_sub(&request_count, 1);

        if (msgq->enable_lock && msgq->unlock_between_callbacks){
            pthread_mutex_lock(&msgq->lock);
		}
    }

    if (msgq->enable_lock){
        pthread_mutex_unlock(&msgq->lock);
	}
}

struct event_msgqueue *msgqueue_new(struct event_base *base, int queue_hwm, int queue_lwm, bool enable_lock, msgqueue_callback callback, void *cbarg) 
{
    struct event_msgqueue *msgq;
    struct circqueue *cq;
    int fds[2];

    cq = circqueue_new(queue_hwm);
    if (cq == NULL){
        return(NULL);
	}

    if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != 0) {
        circqueue_destroy(cq);
        return(NULL);
    }

    if (!(msgq = (struct event_msgqueue*)malloc(sizeof(struct event_msgqueue)))) {
        circqueue_destroy(cq);
        close(fds[0]);
        close(fds[1]);
        return(NULL);
    }

    msgq->push_fd = fds[0];
    msgq->pop_fd = fds[1];
    setnonblock(msgq->pop_fd);

    msgq->queue_lwm = queue_lwm;
    msgq->queue = cq;
    msgq->callback = callback;
    msgq->cbarg = cbarg;
    msgq->evbase = base;
    pthread_mutex_init(&msgq->lock, NULL);
    event_set(&msgq->queue_ev, msgq->pop_fd, EV_READ | EV_PERSIST, msgqueue_pop, msgq);
    event_base_set(base, &msgq->queue_ev);
    event_add(&msgq->queue_ev, NULL);

    msgq->enable_lock = enable_lock;
    msgq->unlock_between_callbacks = 1;

    return(msgq);
}

long get_request_count()
{
    long count;
    count = request_count;
    return count;
}

void msgqueue_destroy(struct event_msgqueue *msgq)
{
    for( ; msgqueue_length(msgq) > 0; ) {
        usleep(100);
    }

    for( ; get_request_count() > 0; ) {
        usleep(100);
    }
}


int msgqueue_push(struct event_msgqueue *msgq, void *msg) 
{
    const char buf[1] = {0};
    int r = -1;

    assert(msgq!=NULL); 
    if (msgq->enable_lock) {
        pthread_mutex_lock(&msgq->lock);
    }
    if ((r = circqueue_push_tail(msgq->queue, msg)) == msgq->queue_lwm) {
        write(msgq->push_fd, buf, 1);
    }

    if (msgq->enable_lock) {
        pthread_mutex_unlock(&msgq->lock);
    }
    return r;
}

unsigned int msgqueue_length(struct event_msgqueue *msgq) 
{
    unsigned int len;

    if (msgq->enable_lock){
        pthread_mutex_lock(&msgq->lock);
	}

    len = circqueue_get_length(msgq->queue);

    if (msgq->enable_lock){
        pthread_mutex_unlock(&msgq->lock);
	}

    return(len);
}
